package com.atguigu.ting.realtime.app.dwd.db;

import com.atguigu.ting.realtime.app.BaseSqlApp;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/11 14:06
 */
public class DwdUserLogin extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdUserLogin().init(
            30051,
            2,
            "DwdUserLogin"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 1. 读取 dwd层页面日志数据
        tEnv.executeSql(
            "create table page_log(" +
                "common map<string, string>, " +
                "page map<string, string>, " +
                "pt as proctime(), " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "DwdUserLogin"));
        // 2. 过滤出登录记录
        Table userLogin = tEnv.sqlQuery(
            "select " +
                "common['uid'] uid, " +
                "ts " +
                "from page_log " +
                "where common['uid'] is not null " +
                "and (" +
                " page['last_page_id'] is null " +
                " or page['last_page_id']='login' " +
                ") ");
        
        // 3. 写出到 kafka 中
        tEnv.executeSql(
            "create table dwd_user_login(" +
                "uid string, " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_USER_LOGIN));
        userLogin.executeInsert("dwd_user_login");
    }
}
